package com.codejiwei.flink.practice;

import com.codejiwei.flink.entity.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author jiwei
 * @description
 * @date 2023/5/24 13:37
 */
public class Flink_ProcessFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(10);

        env.socketTextStream("localhost", 8889)
                .map(line -> {
                    String[] fields = line.split(",");
                    return new WaterSensor(fields[0], Long.valueOf(fields[1]), Integer.valueOf(fields[2]));
                })
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor value, Context context, Collector<String> out) throws Exception {
//                        context.timerService().registerEventTimeTimer(System.currentTimeMillis());
//                        context.timerService().registerProcessingTimeTimer(1111L);

                        String currentKey = context.getCurrentKey();
                        System.out.println("Context current key: " + currentKey);


                        Long timestamp = context.timestamp();
                        System.out.println("Context timestamp: " + timestamp);
                        long currentWatermark = context.timerService().currentWatermark();
                        long currentProcessingTime = context.timerService().currentProcessingTime();
                        System.out.println("Context current watermark: " + currentWatermark);
                        System.out.println("Context current processing time: " + currentProcessingTime);

                        out.collect(value.toString());
                    }
                })
//                .process(new ProcessFunction<WaterSensor, String>() {
//                    @Override
//                    public void processElement(WaterSensor waterSensor, Context context, Collector<String> collector) throws Exception {
//
//                        Long timestamp = context.timestamp();
//                        System.out.println("Context timestamp: " + timestamp);
//                        long currentWatermark = context.timerService().currentWatermark();
//                        long currentProcessingTime = context.timerService().currentProcessingTime();
//                        System.out.println("Context current watermark: " + currentWatermark);
//                        System.out.println("Context current processing time: " + currentProcessingTime);
//
//                        collector.collect(waterSensor.toString());
//                    }
//                })
                .print();

        env.execute();
    }
}
